package org.xxd.kafka.clients.common.network;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.nio.channels.SelectionKey;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-28 09:23
 * @description: 包装了一层的SocketChannel
 * 里面包含一个传输组件，接收组件
 * 还有一个暂存发送数据的组件
 * 通过SelectionKey.attach(KafkaChannel) 就给关联起来了
 */
public class KafkaChannel {
    private static final Logger log = LoggerFactory.getLogger(KafkaChannel.class);

    private final String id;
    private final TransportLayer transportLayer;
    private final int maxReceiveSize;
    private NetworkReceive receive;
    private Send send;

    public KafkaChannel(String id,
                        TransportLayer transportLayer,
                        int maxReceiveSize) throws IOException {
        this.id = id;
        this.transportLayer = transportLayer;
        this.maxReceiveSize = maxReceiveSize;
    }

    /**
     * Does handshake of transportLayer and authentication using configured authenticator
     */
    public void prepare() throws IOException {
        if (!transportLayer.ready())
            transportLayer.handshake();
    }

    public void disconnect() {
        transportLayer.disconnect();
    }

    public boolean finishConnect() throws IOException {
        return transportLayer.finishConnect();
    }

    public boolean isConnected() {
        return transportLayer.isConnected();
    }

    public String id() {
        return id;
    }

    /**
     * socket通道变成哑子(听不到声音，就接收不到数据)
     */
    public void mute() {
        transportLayer.removeInterestOps(SelectionKey.OP_READ);
    }

    public void unmute() {
        transportLayer.addInterestOps(SelectionKey.OP_READ);
    }

    public boolean isMute() {
        return transportLayer.isMute();
    }

    public boolean ready() {
        return transportLayer.ready();
    }

    public boolean hasSend() {
        return send != null;
    }

    public InetAddress socketAddress() {
        return transportLayer.socketChannel().socket().getInetAddress();
    }

    public void setSend(Send send) {
        if (this.send != null)
            throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
        this.send = send;
        log.info(" {} 新增对【OP_WRITE】事件的关注", transportLayer.socketChannel());

        this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
    }

    /**
     * 接收到数据，读取出来
     * @return
     * @throws IOException
     */
    public NetworkReceive read() throws IOException {
        NetworkReceive result = null;

        if (receive == null) {
            receive = new NetworkReceive(maxReceiveSize, id);
        }

        receive(receive);
        if (receive.complete()) {
            receive.payload().rewind();
            result = receive;
            receive = null;
        }
        return result;
    }

    /**
     * 写数据
     * @return
     * @throws IOException
     */
    public Send write() throws IOException {
        Send result = null;
        if (send != null && send(send)) {
            result = send;
            send = null;
        }
        return result;
    }

    private long receive(NetworkReceive receive) throws IOException {
        return receive.readFrom(transportLayer);
    }

    private boolean send(Send send) throws IOException {
        send.writeTo(transportLayer);
        if (send.completed()) {
            log.info("{} 发送完毕，取消对write事件的关注", transportLayer.socketChannel());
            transportLayer.removeInterestOps(SelectionKey.OP_WRITE);
        }

        return send.completed();
    }
}
